package com.xnyzc.lhy.order.mq.consumer;

import com.alibaba.fastjson.JSON;
import com.xnyzc.lhy.common.component.config.security.RocketConfig;
import com.xnyzc.lhy.common.component.redis.RedisCommon;
import com.xnyzc.lhy.common.constant.MqConstant;
import com.xnyzc.lhy.common.constant.MqConsumerGroupConstant;
import com.xnyzc.lhy.common.entity.DiDiRv;
import com.xnyzc.lhy.common.entity.Rv;
import com.xnyzc.lhy.common.penum.EErrorCode;
import com.xnyzc.lhy.common.util.CheckUtil;
import com.xnyzc.lhy.order.entity.param.synchronization.StatusCallbackParam;
import com.xnyzc.lhy.order.feign.didi.IDidiService;
import lombok.extern.slf4j.Slf4j;
import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyContext;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyStatus;
import org.apache.rocketmq.client.consumer.listener.MessageListenerConcurrently;
import org.apache.rocketmq.common.consumer.ConsumeFromWhere;
import org.apache.rocketmq.common.message.MessageExt;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.stereotype.Component;

import javax.annotation.PostConstruct;
import java.util.List;

/**
 * 司机状态同步
 *
 * @author c
 */
@Slf4j
@Component
public class MqStatusCallBackSyncConsumer {

    private DefaultMQPushConsumer consumer;

    @Autowired
    RocketConfig config;

    @Autowired
    RedisCommon redisCommon;

    @Value("${rocketMq.isStart}")
    private boolean isStart;

    @Autowired
    private IDidiService iDidiService;

    /**
     * 开启监听
     */
    @PostConstruct
    public void init() {
        try {
            if (!isStart) {
                return;
            }
            consumer = new DefaultMQPushConsumer(MqConsumerGroupConstant.STATUS_CALL_BACK);
            consumer.setNamesrvAddr(config.getNameServerAddr());
            consumer.subscribe(config.getTopic(), MqConstant.STATUS_CALL_BACK_SYNC);
            consumer.setConsumeMessageBatchMaxSize(1);
            consumer.setConsumeFromWhere(ConsumeFromWhere.CONSUME_FROM_LAST_OFFSET);
            consumer.registerMessageListener(new MessageListenerConcurrently() {
                @Override
                public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> msgs, ConsumeConcurrentlyContext context) {
                    MessageExt messageExt = msgs.get(0);
                    return MqStatusCallBackSyncConsumer.this.dealBody(messageExt);
                }
            });
            consumer.start();
            log.info("---------司机状态同步消费者启动完成----o(*^▽^*)┛----");
        } catch (Exception e) {
            log.info("---------司机状态同步消费者初始化失败---(# ﾟДﾟ)--------");
            log.error(e.getMessage());
        }
    }

    /**
     * 处理业务逻辑
     *
     * @param messageExt
     * @return
     */
    public ConsumeConcurrentlyStatus dealBody(MessageExt messageExt) {
        StatusCallbackParam statusCallbackParam = null;

        try {
            //发送的消息
            String body = new String(messageExt.getBody());
            if (CheckUtil.strIsEmpty(body)) {
                return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
            }
            statusCallbackParam = JSON.parseObject(body, StatusCallbackParam.class);

            statusCallbackParam.setIsRetry(true);

            log.info("ORDER_HISTORY 订单号:[{}],向滴滴同步订单状态,请求参数:[{}],调用司机状态到滴滴请求参数:[{}]", statusCallbackParam.getOrder_id(), statusCallbackParam.getStatus(), statusCallbackParam);
            // 推送给滴滴状态
            Rv rv = iDidiService.statusCallback(statusCallbackParam);
            if (rv.getCode() == EErrorCode.success.getValue()) {
                log.info("ORDER_HISTORY 订单号:[{}],向滴滴同步订单状态,请求参数:[{}],成功", statusCallbackParam.getOrder_id(), statusCallbackParam);
            } else {
                log.info("ORDER_HISTORY 订单号:[{}],向滴滴同步订单状态,请求参数:[{}],失败返回值:[{}]", statusCallbackParam.getOrder_id(), rv);
            }

            return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
        } catch (Exception e) {
            log.error(e.getMessage());
        }
        return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
    }
}
